package com.atguigu.flink08;

import com.atguigu.beans.WaterSensor;
import com.atguigu.func.WaterSensorMapFunction;
import org.apache.flink.api.common.state.MapState;
import org.apache.flink.api.common.state.MapStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.util.Collector;

import java.util.Map;

/**
 * @author Felix
 * @date 2024/2/27
 * 该案例演示Map状态
 * 需求：统计每种传感器每种水位值出现的次数
 */
public class Flink05_State_MapState {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        SingleOutputStreamOperator<WaterSensor> wsDS = env
                .socketTextStream("hadoop102", 8888)
                .map(new WaterSensorMapFunction());

        wsDS
                .keyBy(WaterSensor::getId)
                .process(
                        new KeyedProcessFunction<String, WaterSensor, String>() {
                            //声明状态
                            MapState<Integer,Integer> vcCountMapState;

                            @Override
                            public void open(Configuration parameters) throws Exception {
                                //状态初始化
                                MapStateDescriptor<Integer, Integer> mapStateDescriptor
                                        = new MapStateDescriptor<Integer, Integer>("vcCountMapState", Integer.class, Integer.class);
                                vcCountMapState = getRuntimeContext().getMapState(mapStateDescriptor);
                            }

                            @Override
                            public void processElement(WaterSensor ws, KeyedProcessFunction<String, WaterSensor, String>.Context ctx, Collector<String> out) throws Exception {
                                Integer vc = ws.vc;
                                //判断状态中是否包含当前水位
                                if(vcCountMapState.contains(vc)){
                                    vcCountMapState.put(vc,vcCountMapState.get(vc) + 1);
                                }else{
                                    vcCountMapState.put(vc,1);
                                }

                                StringBuilder str = new StringBuilder();
                                str.append("==================\n");
                                str.append("当前传感器 " +ctx.getCurrentKey()+ ",各个水位值出现次数如下: \n");
                                for (Map.Entry<Integer, Integer> entry : vcCountMapState.entries()) {
                                    str.append("水位值:"+entry.getKey()+"出现次数:"+ entry.getValue() + "\n");
                                }
                                str.append("==================\n");

                                out.collect(str.toString());

                            }
                        }
                ).print();

        env.execute();
    }
}
